#include <errno.h>
#include <pthread.h>
#include <string.h>
#include <unistd.h>

#include "compat.h"

#include "psd_logger.h"
#include "psd_measurement.h"
#include "psd_session.h"
#include "psd_xfer.h"
#include "psd_backend_slabs.h"

#define SEND_SIZE (64*1024)

void *psd_slabs_ingress_xfer_thread(void *arg) {
        psdXferArgs *xfer_args = arg;
	psdConn *curr_src = xfer_args->src;
	psdConn *curr_sink = xfer_args->sink;
	unsigned long slab_free_bytes;
	unsigned long bytes_read;
	size_t bytes_read_size;
	struct timeval start_time, end_time;
	char filename[255];

	pthread_mutex_lock(&(xfer_args->lock));

	psd_info(2, "new transfer thread: %s->%s", curr_src->description, curr_sink->description);

	gettimeofday(&start_time, NULL);

	// alert the parent that we're running and about to wait for them to tell us how much data to transfer
	pthread_cond_signal(&(xfer_args->cond)); 
	
	while (1) {
	        pthread_cond_wait(&(xfer_args->cond), &(xfer_args->lock));

		if (xfer_args->xfer_amt == (uint64_t) -1) {
			psd_info(2, "%s->%s: going into perpetual receive mode", curr_src->description, curr_sink->description);
		} else {
			psd_info(2, "%s->%s: receiving %llu bytes", curr_src->description, curr_sink->description, xfer_args->xfer_amt);
		}

		bytes_read = 0;

		while(bytes_read < xfer_args->xfer_amt) {
			int recv_amt;
			int n;
			char *slab_buf_ptr;

			// doesn't seem the connection can get frozen, so this seems like unnecessary overhead
			/*
			pthread_mutex_lock(&(curr_src->lock));
			{
				while (curr_src->status == STATUS_FROZEN) {
					psd_info(0, "%s->%s: connection %s frozen", curr_src->description, curr_sink->description, curr_src->description);
					pthread_cond_wait(&(curr_src->cond), &(curr_src->lock));
				}
			}
			pthread_mutex_unlock(&(curr_src->lock));
			*/

			pthread_mutex_lock(&(xfer_args->buf_lock));
			{
				slab_free_bytes = ring_buffer_count_free_bytes(xfer_args->buf);;
				
				if (slab_free_bytes == 0) {
					psd_info(5, "%s->%s: slab buffer is full, waiting to write", curr_src->description, curr_sink->description);
					pthread_cond_wait(&(xfer_args->buf_cond), &(xfer_args->buf_lock));
					slab_free_bytes = ring_buffer_count_free_bytes(xfer_args->buf);
				}

				if (PSD_BUF_SIZE < slab_free_bytes) {
					recv_amt = PSD_BUF_SIZE;
				}
				else {
					recv_amt = slab_free_bytes;
				}

				psd_info(11, "%s->%s: going into recv with %lu free slab bytes",
					 curr_src->description, curr_sink->description, slab_free_bytes);
				
				slab_buf_ptr = ring_buffer_write_address(xfer_args->buf);
				n = psd_conn_read(curr_src, slab_buf_ptr, recv_amt, 0);
		      
				psd_info(11, "%s->%s: recv returned %d", curr_src->description, curr_sink->description, n);		      
				
				if (n <= 0) {
					if(n < 0) {
						psd_err(0, "%s->%s: recv failed: %s", curr_src->description, curr_sink->description, strerror(errno));
						goto out;
					}
					
					if (ring_buffer_count_bytes(xfer_args->buf) > 0)
						// need to send this buffer before closing...
						goto out;
					else
						goto out;
				}
				
				ring_buffer_write_advance(xfer_args->buf, n);
				bytes_read += n;
			}
			pthread_mutex_unlock(&(xfer_args->buf_lock));
			
			pthread_mutex_lock(&xfer_total_lock);
			{
				xfer_total_bytes += n;
				if (xfer_total_bytes >= PSD_SLAB_SIZE)
					pthread_cond_signal(&xfer_total_cond);
			}
			pthread_mutex_unlock(&xfer_total_lock);
		}
			
		if (write(xfer_args->parent_fd, "1", 1) < 0) {
			psd_err(0, "%s->%s: couldn't communicate with parent thread: %s", curr_src->description, curr_sink->description, strerror(errno));
			goto out;
		}
	}
	
out:
	psd_info(1, "%s->%s: closing sockets and returning.", curr_src->description, curr_sink->description);
	
	gettimeofday(&end_time, NULL);
	
	bytes_read_size = sizeof(bytes_read);
	if (psd_conn_get_stat(curr_sink, PSD_STATS_BYTES_WRITTEN, &bytes_read, &bytes_read_size) == 0) {
		char tmp[255];
		snprintf(tmp, sizeof(tmp), "%s/%s->%s", psd_session_get_id(xfer_args->sess), curr_src->description, curr_sink->description);
		psd_log_measurement_dbl(end_time, tmp, "TransferRate", ((bytes_read/difftv(&start_time, &end_time))*8)/1000/1000);
		psd_log_measurement_uint(end_time, tmp, "TransferBytes", bytes_read);
		psd_log_measurement_dbl(end_time, tmp, "TransferTime", difftv(&start_time, &end_time));
	}
	
	/*
	  fclose(output);
	*/

	psd_conn_shutdown(curr_sink, PSD_SEND_SIDE);
	psd_conn_shutdown(curr_src, PSD_RECV_SIDE);
	
	pthread_exit(NULL);
}

void *psd_slabs_egress_xfer_thread(void *arg) {
        psdXferArgs *xfer_args = arg;
        psdConn *curr_src = xfer_args->src;
        psdConn *curr_sink = xfer_args->sink;
        unsigned long slab_bytes;
        unsigned long bytes_sent;
        size_t bytes_sent_size;
        struct timeval start_time, end_time;
        char filename[255];

        pthread_mutex_lock(&(xfer_args->lock));

        psd_info(2, "new transfer thread: %s->%s", curr_src->description, curr_sink->description);

        gettimeofday(&start_time, NULL);

        // alert the parent that we're running and about to wait for them to tell us how much data to transfer
        pthread_cond_signal(&(xfer_args->cond));

        while (1) {
                pthread_cond_wait(&(xfer_args->cond), &(xfer_args->lock));

                if (xfer_args->xfer_amt == (uint64_t) -1) {
                        psd_info(2, "%s->%s: going into perpetual receive mode", curr_src->description, curr_sink->description);
                } else {
                        psd_info(2, "%s->%s: receiving %llu bytes", curr_src->description, curr_sink->description, xfer_args->xfer_amt);
                }

		bytes_sent = 0;

		while (bytes_sent < xfer_args->xfer_amt) {
			int send_amt;
			int n;
			char *slab_buf_ptr;

			// doesn't seem the connection can get frozen, so this seems like unnecessary overhead
			/*
			pthread_mutex_lock(&(curr_sink->lock));
			{
				while (curr_sink->status == STATUS_FROZEN) {
					psd_info(0, "%s->%s: connection %s frozen", curr_src->description, curr_sink->description, curr_sink->description);
					pthread_cond_wait(&(curr_sink->cond), &(curr_sink->lock));
				}
			}
			pthread_mutex_unlock(&(curr_sink->lock));
			*/
			
			pthread_mutex_lock(&(xfer_args->buf_lock));
			{

				while (1) {
					slab_bytes = ring_buffer_count_bytes(xfer_args->buf);

					if (slab_bytes <= 0) {
						psd_info(5, "%s->%s: slab buffer is empty, waiting to read",
							 curr_src->description, curr_sink->description);
						pthread_cond_wait(&(xfer_args->buf_cond), &(xfer_args->buf_lock));
						slab_bytes = ring_buffer_count_bytes(xfer_args->buf);
					}
					
					if (PSD_BUF_SIZE < slab_bytes) {
						send_amt = PSD_BUF_SIZE;
					}
					else {
						send_amt = slab_bytes;
					}

					slab_buf_ptr = ring_buffer_read_address(xfer_args->buf);

					psd_info(11, "%s->%s: going into send (%lu) with %lu slab bytes remaining",
						 curr_src->description, curr_sink->description, send_amt, slab_bytes);
										
					
					n = psd_conn_write(curr_sink, slab_buf_ptr, send_amt, 0);
					
					if (n < 0) {
						psd_err(0, "%s->%s: send failed: %s", curr_src->description, curr_sink->description, strerror(errno));
						goto out;
					}
					psd_info(11, "%s->%s: send returned %d", curr_src->description, curr_sink->description, n);
					
					ring_buffer_read_advance(xfer_args->buf, n);
					bytes_sent += n;
				}
			}
			pthread_mutex_unlock(&(xfer_args->buf_lock));
		}
		
		if (write(xfer_args->parent_fd, "1", 1) < 0) {
			psd_err(0, "%s->%s: couldn't communicate with parent thread: %s", curr_src->description, curr_sink->description, strerror(errno));
			goto out;
		}
        }
	
 out:
        psd_info(1, "%s->%s: closing sockets and returning.", curr_src->description, curr_sink->description);
	
        gettimeofday(&end_time, NULL);
	
        bytes_sent_size = sizeof(bytes_sent);
        if (psd_conn_get_stat(curr_sink, PSD_STATS_BYTES_WRITTEN, &bytes_sent, &bytes_sent_size) == 0) {
                char tmp[255];
                snprintf(tmp, sizeof(tmp), "%s/%s->%s", psd_session_get_id(xfer_args->sess), curr_src->description, curr_sink->description);
                psd_log_measurement_dbl(end_time, tmp, "TransferRate", ((bytes_sent/difftv(&start_time, &end_time))*8)/1000/1000);
                psd_log_measurement_uint(end_time, tmp, "TransferBytes", bytes_sent);
                psd_log_measurement_dbl(end_time, tmp, "TransferTime", difftv(&start_time, &end_time));
        }

	/*
        fclose(output);
        */

        psd_conn_shutdown(curr_sink, PSD_SEND_SIDE);
        psd_conn_shutdown(curr_src, PSD_RECV_SIDE);

        pthread_exit(NULL);
}
